package com.atguigu.realtime.app.dwm;

import com.alibaba.fastjson.JSON;
import com.alibaba.fastjson.JSONObject;
import com.atguigu.realtime.app.BaseAppV2;
import com.atguigu.realtime.bean.OrderDetail;
import com.atguigu.realtime.bean.OrderInfo;
import com.atguigu.realtime.bean.OrderWide;
import com.atguigu.realtime.util.*;
import org.apache.flink.api.common.eventtime.WatermarkStrategy;
import org.apache.flink.configuration.Configuration;
import org.apache.flink.streaming.api.datastream.AsyncDataStream;
import org.apache.flink.streaming.api.datastream.DataStreamSource;
import org.apache.flink.streaming.api.datastream.KeyedStream;
import org.apache.flink.streaming.api.datastream.SingleOutputStreamOperator;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.flink.streaming.api.functions.async.ResultFuture;
import org.apache.flink.streaming.api.functions.async.RichAsyncFunction;
import org.apache.flink.streaming.api.functions.co.ProcessJoinFunction;
import org.apache.flink.streaming.api.windowing.time.Time;
import org.apache.flink.util.Collector;
import redis.clients.jedis.Jedis;

import java.sql.Connection;
import java.time.Duration;
import java.util.Collections;
import java.util.Map;
import java.util.concurrent.ThreadPoolExecutor;
import java.util.concurrent.TimeUnit;

import static com.atguigu.realtime.common.Constant.PHOENIX_DRIVER;
import static com.atguigu.realtime.common.Constant.PHOENIX_URL;

/**
 * @Author lizhenchao@atguigu.cn
 * @Date 2021/4/20 14:22
 */
public class DWMOrderWideApp_Cache_Async extends BaseAppV2 {
    public static void main(String[] args) {
        new DWMOrderWideApp_Cache_Async().init(3003,
                                               2,
                                               "DWMOrderWideApp_Cache",
                                               "DWMOrderWideApp_Cache",
                                               "dwd_order_info", "dwd_order_detail");
    }
    
    @Override
    protected void run(StreamExecutionEnvironment env,
                       Map<String, DataStreamSource<String>> streams) {
        // 1. 事实表join
        SingleOutputStreamOperator<OrderWide> orderWideStreamWithoutDim = factTableJoin(streams);
        // 2. join 维度(补齐维度信息)
        SingleOutputStreamOperator<OrderWide> orderWideStreamWithDim = joinDim(orderWideStreamWithoutDim);
        // 3.把order_wide的宽表数据写入不到kafka
        send2Kafka(orderWideStreamWithDim);
        
    }
    
    private void send2Kafka(SingleOutputStreamOperator<OrderWide> orderWideStreamWithDim) {
        orderWideStreamWithDim
            .map(JSON::toJSONString)
            .addSink(MyKafkaUtil.getKafkaSink("dwm_order_wide"));
    }
    
    private SingleOutputStreamOperator<OrderWide> joinDim(SingleOutputStreamOperator<OrderWide> orderWideStreamWithoutDim) {
        
        SingleOutputStreamOperator<OrderWide> result = AsyncDataStream
            .unorderedWait(orderWideStreamWithoutDim,
                           new RichAsyncFunction<OrderWide, OrderWide>() {
                               private ThreadPoolExecutor threadPool;
                               private Connection conn;
                
                               @Override
                               public void open(Configuration parameters) throws Exception {
                                   conn = MyJdbcUtil.getJdbcConnection(PHOENIX_URL, PHOENIX_DRIVER);
                                   threadPool = MyThreadPollUtil.getThreadPool();
                               }
                
                               @Override
                               public void asyncInvoke(OrderWide orderWide,
                                                       ResultFuture<OrderWide> resultFuture) throws Exception {
                                   // 异步调用函数
                                   threadPool.execute(new Runnable() {
                                       @Override
                                       public void run() {
                                           Jedis jedis = MyRedisUtil.getClient();
                                           // 1. join 用户维度, 根据用户id去查询对应的用户的信息
                                           JSONObject user = MyDimUtil.readDim(conn, jedis, "dim_user_info", orderWide.getUser_id());
                                           orderWide.calcUserAge(user.getString("BIRTHDAY"));
                                           orderWide.setUser_gender(user.getString("GENDER"));
                                           // 2. join 省份维度
                            
                                           final JSONObject provinceObj = MyDimUtil.readDim(conn, jedis, "dim_base_province", orderWide.getProvince_id());
                                           orderWide.setProvince_3166_2_code(provinceObj.getString("ISO_3166_2"));
                                           orderWide.setProvince_area_code(provinceObj.getString("AREA_CODE"));
                                           orderWide.setProvince_name(provinceObj.getString("NAME"));
                                           orderWide.setProvince_iso_code(provinceObj.getString("ISO_CODE"));
                            
                                           // 3. join sku维度
                                           final JSONObject skuObj = MyDimUtil.readDim(conn, jedis, "dim_sku_info", orderWide.getSku_id());
                                           orderWide.setSku_name(skuObj.getString("SKU_NAME"));
                                           orderWide.setSpu_id(skuObj.getLong("SPU_ID"));
                                           orderWide.setCategory3_id(skuObj.getLong("CATEGORY3_ID"));
                                           orderWide.setTm_id(skuObj.getLong("TM_ID"));
                            
                                           // 4. join spu维度
                                           final JSONObject spuObj = MyDimUtil.readDim(conn, jedis, "dim_spu_info", orderWide.getSpu_id());
                                           orderWide.setSpu_name(spuObj.getString("SPU_NAME"));
                            
                                           // 5. join category3维度
                                           final JSONObject category3Obj = MyDimUtil.readDim(conn, jedis, "dim_base_category3", orderWide.getCategory3_id());
                                           orderWide.setCategory3_name(category3Obj.getString("NAME"));
                                           // 6. join 品牌维度
                                           final JSONObject tmObj = MyDimUtil.readDim(conn, jedis, "dim_base_trademark", orderWide.getTm_id());
                                           orderWide.setTm_name(tmObj.getString("TM_NAME"));
                            
                                           jedis.close();
                                           resultFuture.complete(Collections.singletonList(orderWide));
                                       }
                                   });
                    
                               }
                
                               @Override
                               public void close() throws Exception {
                                   if (conn != null) {
                                       conn.close();
                                   }
                    
                                   if (threadPool != null) {
                                       threadPool.shutdown();
                                   }
                               }
                           },
                           30,
                           TimeUnit.SECONDS);
        
        return result;
        
    }
    
    private SingleOutputStreamOperator<OrderWide> factTableJoin(Map<String, DataStreamSource<String>> streams) {
        // interval join:  1. 必须是keyBy 2. 必须是事件时间
        KeyedStream<OrderInfo, Long> orderInfoStream = streams
            .get("dwd_order_info")
            .map(jsonString -> JSON.parseObject(jsonString, OrderInfo.class))
            .assignTimestampsAndWatermarks(
                WatermarkStrategy
                    .<OrderInfo>forBoundedOutOfOrderness(Duration.ofSeconds(2))
                    .withTimestampAssigner((orderInfo, ts) -> orderInfo.getCreate_ts())
            )
            .keyBy(OrderInfo::getId);
        
        KeyedStream<OrderDetail, Long> orderDetailStream = streams
            .get("dwd_order_detail")
            .map(jsonString -> JSON.parseObject(jsonString, OrderDetail.class))
            .assignTimestampsAndWatermarks(
                WatermarkStrategy
                    .<OrderDetail>forBoundedOutOfOrderness(Duration.ofSeconds(2))
                    .withTimestampAssigner((orderDetail, ts) -> orderDetail.getCreate_ts())
            )
            .keyBy(OrderDetail::getOrder_id);
        
        return orderInfoStream
            .intervalJoin(orderDetailStream)
            .between(Time.seconds(-5), Time.seconds(5))
            .process(new ProcessJoinFunction<OrderInfo, OrderDetail, OrderWide>() {
                @Override
                public void processElement(OrderInfo orderInfo,
                                           OrderDetail orderDetail,
                                           Context ctx,
                                           Collector<OrderWide> out) throws Exception {
                    out.collect(new OrderWide(orderInfo, orderDetail));
                }
            });
        
    }
}
/*
把重复使用的数据存入到redis
维度信息:
    6张维度表的数据
    
string list set hash zset

用户维度
string ?
 key                    value
 "dim_user_info":1      "json格式字符串"
 
 优点:
   1. 读写方便
   2. 可以单独给每个数据设置过期时间
 缺点:
    key的数量庞大
    
list set ?
    "dim_user_info"         List["json格式数据", ""]
    缺点: 取的时候及其不方便
    
hash ?
    key                 value(map)
    "dim_user_info"      field      value
                          id_1      "json格式字符串"
                          
    优点: 一个维度表一个key
    缺点: 没有办法单独给每条数据设置过期时间

 */